package com.up.kafka.config;

import com.up.kafka.productor.ByteArrayMessageProducer;
import com.up.kafka.productor.KafkaProducer;
import com.up.kafka.productor.ProducerBeanName;
import com.up.kafka.productor.StringMessageProducer;
import com.up.kafka.sender.ByteArraySender;
import com.up.kafka.sender.KryoArraySender;
import com.up.kafka.sender.MessageSender;
import com.up.kafka.sender.StringSender;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;

@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(KafkaTemplate.class)
@ConditionalOnProperty(prefix = "kafka", value = "enable", havingValue = "true")
@AutoConfigureAfter(KafkaAutoConfiguration.class)
public class KafkaAssemblyAutoConfiguration {

    @Bean
    @ConditionalOnClass(KafkaTemplate.class)
    @ConditionalOnBean(KafkaTemplate.class)
    public KafkaProducer<byte[]> byteArrayMessageProducer(KafkaTemplate<String, byte[]> kafkaTemplate) {
        return new ByteArrayMessageProducer(kafkaTemplate);
    }

    @Bean
    @ConditionalOnClass(KafkaTemplate.class)
    @ConditionalOnBean(KafkaTemplate.class)
    public KafkaProducer<String> stringMessageProducer(KafkaTemplate<String, String> kafkaTemplate) {
        return new StringMessageProducer(kafkaTemplate);
    }

    @Bean(ProducerBeanName.BYTE_ARRAY_SENDER_BEAN_NAME)
    @ConditionalOnBean(KafkaProducer.class)
    public MessageSender byteArraySender(KafkaProducer<byte[]> byteArrayMessageProducer) {
        return new ByteArraySender(byteArrayMessageProducer);
    }

    @Bean(ProducerBeanName.KRYO_ARRAY_SENDER_BEAN_NAME)
    @ConditionalOnBean(KafkaProducer.class)
    public MessageSender kryoArraySender(KafkaProducer<byte[]> byteArrayMessageProducer) {
        return new KryoArraySender(byteArrayMessageProducer);
    }

    @Bean(ProducerBeanName.STRING_SENDER_BEAN_NAME)
    @ConditionalOnBean(KafkaProducer.class)
    public MessageSender stringSender(KafkaProducer<String> stringMessageProducer) {
        return new StringSender(stringMessageProducer);
    }
}
